package driver

import (
	"context"
	"encoding/json"

	"task/ecode"
	"task/models"

	"github.com/apache/pulsar-client-go/pulsar"
	log "github.com/sirupsen/logrus"
)

// SendDispatchEventMsg send dispatch event msg.
func SendDispatchEventMsg(ctx context.Context, event *models.DispatchEvent) error {
	var err error
	msg, _ := json.Marshal(event)
	if _, err = DispatchProducer.Send(ctx, &pulsar.ProducerMessage{Payload: msg}); err != nil {
		log.Errorf("SendDispatchEventMsg send msg error(%v)", err)
		return ecode.PulsarError
	}
	return nil
}

// SendExecuteEventMsg send execute event msg.
func SendExecuteEventMsg(ctx context.Context, events []*models.ProcessEvent) error {
	var err error
	for _, event := range events {
		msg, _ := json.Marshal(event)
		if _, err = ExecuteProducer.Send(ctx, &pulsar.ProducerMessage{Payload: msg}); err != nil {
			log.Errorf("SendExecuteEventMsg send msg error(%v)", err)
			return ecode.PulsarError
		}
	}
	return nil
}

// SendBusiness1EventMsg send business1 event msg.
func SendBusiness1EventMsg(ctx context.Context, event *models.BusinessEvent) error {
	var err error
	msg, _ := json.Marshal(event)
	if _, err = Business1Producer.Send(ctx, &pulsar.ProducerMessage{Payload: msg}); err != nil {
		log.Errorf("SendBusiness1EventMsg send msg error(%v)", err)
		return ecode.PulsarError
	}
	return nil
}

// SendBusiness2EventMsg send business2 event msg.
func SendBusiness2EventMsg(ctx context.Context, event *models.BusinessEvent) error {
	var err error
	msg, _ := json.Marshal(event)
	if _, err = Business2Producer.Send(ctx, &pulsar.ProducerMessage{Payload: msg}); err != nil {
		log.Errorf("SendBusiness2EventMsg send msg error(%v)", err)
		return ecode.PulsarError
	}
	return nil
}

// SendCallbackEventMsg send callback event msg.
func SendCallbackEventMsg(ctx context.Context, event *models.CallbackEvent) error {
	var err error
	msg, _ := json.Marshal(event)
	if _, err = CallbackProducer.Send(ctx, &pulsar.ProducerMessage{Payload: msg}); err != nil {
		log.Errorf("SendCallbackEventMsg send msg error(%v)", err)
		return ecode.PulsarError
	}
	return nil
}
